package com.example.demo2.common.util.asyncmessage.client;

import java.io.UnsupportedEncodingException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.internal.security.SSLSocketFactoryFactory;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class PahoAsyncMessageClient implements MqttCallback {
	public static final String KEYSTORE_TYPE = SSLSocketFactoryFactory.KEYSTORETYPE;
	public static final String TRUSTSTORE_TYPE = SSLSocketFactoryFactory.TRUSTSTORETYPE;
	public static final String KEYSTORE_PASSWORD = SSLSocketFactoryFactory.KEYSTOREPWD;
	public static final String KEYSTORE_PATH = SSLSocketFactoryFactory.KEYSTORE;
	public static final String TRUSTSTORE_PATH = SSLSocketFactoryFactory.TRUSTSTORE;
	public static final String TRUSTSTORE_PASSWORD = SSLSocketFactoryFactory.TRUSTSTOREPWD;
	
	private final static short KEEP_ALIVE = 20;// 低耗网络，但是又需要及时获取数据，心跳30s
	private MqttClient mqttClient;
	private MqttConnectOptions options = new MqttConnectOptions();
	private Map<String, MqttMessageListener> listeners;
	

	
	private static PahoAsyncMessageClient instance;
	
	
	public static void init(String ip, int port, SSLConfig sslConfig, String clientId) {
		synchronized (PahoAsyncMessageClient.class) {
			if (instance == null) {
				instance = new PahoAsyncMessageClient(ip, port, null, clientId);
			}
		}
	}
	
	public static PahoAsyncMessageClient getInstance() {
		return instance;
	}

	private PahoAsyncMessageClient(String ip, int port, SSLConfig sslConfig, String clientId) {
		try {
//	    	String tmpDir = System.getProperty("java.io.tmpdir");
//	    	MqttClientPersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
			String scheme = sslConfig != null ? "ssl://" : "tcp://";
			String uri = scheme+ip+":"+port;
			mqttClient = new MqttClient(uri, clientId, new MemoryPersistence());
			mqttClient.setCallback(this);
			options.setServerURIs(new String[]{uri});
			options.setKeepAliveInterval(KEEP_ALIVE);
			options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
			listeners  = new ConcurrentHashMap<String, MqttMessageListener>();
			
			setSSLConfig(sslConfig);

		} catch (MqttException e) {
			throw new IllegalArgumentException(e.getMessage());
		}
	}
	public void setUserName(String userName) {
		options.setUserName(userName);
	}
	
	private void setSSLConfig(SSLConfig config) {
		if (config == null) {
			options.setSSLProperties(null);
		} else {
			options.setSSLProperties(config.toProperties());
		}
	}

	public void setPassword(String password) {
		if (password != null) {
			options.setPassword(password.toCharArray());
		} else {
			options.setPassword(null);
		}
	}
	
	public boolean isConnected() {
		return mqttClient.isConnected();
	}
	
	public synchronized void connect() throws MqttSecurityException, MqttException {
		options.setCleanSession(true);
		if (mqttClient.isConnected()) {
			mqttClient.disconnect();
		}
//		mqttClient.connect(options);
		mqttClient.connectWithResult(options);
	}
	
	public void close() throws MqttException {
		mqttClient.close();
	}
	
	public synchronized void disconnect() throws MqttException {
		mqttClient.disconnect();
	}
	
	public static enum MessageType {
		/**
		 * 私有广播消息，即时消息，服务器不负责缓存
		 */
		PRIVATE_BROADCAST(0, true, false, false),
		/**
		 * 公有广播消息，即时消息，服务器不负责缓存
		 */
		PUBLIC_BROADCAST(0, false, false, false),
		/**
		 * 交易消息，用户消息，服务器负责缓存
		 */
		TRANSACTION(2, true, true, false),
		/**
		 * 私有用户消息，服务器负责缓存
		 */
		PRIVATE_MESSAGE(1, true, true, false),
		/**
		 * 一般操作应答消息，服务器不负责缓存
		 */
		COMMON_RESPONSE(0, true, false, false),
		/**
		 * 系统通知，服务器负责缓存最后一条消息
		 */
		SYSTEM_NOTIFACTION(1, false, false, true),
		;
		
		private int qos;
		private boolean isPrivate;
		private boolean persistent;
		private boolean retainabled;

		MessageType(int qos, boolean isPrivate, boolean persistent, boolean retainabled) {
			this.qos = qos;
			this.isPrivate = isPrivate;
			this.persistent = persistent;
			this.retainabled = retainabled;
		}

		public int getQos() {
			return qos;
		}

		public void setQos(int qos) {
			this.qos = qos;
		}

		public boolean isPrivate() {
			return isPrivate;
		}

		public void setPrivate(boolean isPrivate) {
			this.isPrivate = isPrivate;
		}

		public boolean isPersistent() {
			return persistent;
		}

		public void setPersistent(boolean persistent) {
			this.persistent = persistent;
		}

		public boolean isRetainabled() {
			return retainabled;
		}

		public void setRetainabled(boolean retainabled) {
			this.retainabled = retainabled;
		}
	}
	
	private void publish(String topic, String content, int qos, boolean retained) throws MqttPersistenceException, MqttException {
		try {
			MqttMessage message = new MqttMessage(content.getBytes("utf-8"));
			message.setQos(qos);
			message.setRetained(retained);
			mqttClient.publish(topic, message);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}
	
	public synchronized void publish(String topic, String message, MessageType type) throws MqttPersistenceException, MqttException {
		if (topic == null) {
			throw new IllegalArgumentException("message type is not valid");
		}
		
		publish(topic, message, type.getQos(), type.isRetainabled());
	}
	
	@Override
	public void connectionLost(Throwable cause) {
		System.out.println("connectionLost");
	}
	private Executor executor = Executors.newCachedThreadPool();
	@Override
	public void messageArrived(String topic, final MqttMessage message)
			throws Exception {
		final MqttMessageListener listener = listeners.get(topic);
		if(listener!=null){
			executor.execute(new Runnable() {
				
				@Override
				public void run() {
					listener.onMessage(message);
					
				}
			});
		}
	}
	
	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		System.out.println("deliveryComplete");
	}

	public static class SSLConfig {
		private Properties properties = new Properties();
		public void setKeyStoreType(String type) {
			properties.put(KEYSTORE_TYPE, type);
		}
		
		public void setTrustStoreType(String type) {
			properties.put(TRUSTSTORE_TYPE, type);
		}
		
		public void setKeyStorePassword(String password) {
			properties.put(KEYSTORE_PASSWORD, password);
		}
		
		public void setTrustStorePassword(String password) {
			properties.put(TRUSTSTORE_PASSWORD, password);
		}
		
		public void setKeyStore(String path) {
			properties.put(KEYSTORE_PATH, path);
		}
		
		public void setTrustStore(String path) {
			properties.put(TRUSTSTORE_PATH, path);
		}
		
		Properties toProperties() {
			return properties;
		}
	}
	
	public void addMessageListener(String topic,MqttMessageListener listener) throws MqttException{
		topic = topic.replace('.', '/');
		mqttClient.subscribe(topic);
		listeners.put(topic, listener);
	}
	
}